package cn.me.alphamq.nameserver.event.spi.listener.service;

import cn.me.alphamq.common.entity.resp.PullBrokerAddressResp;
import cn.me.alphamq.common.enumeration.BrokerRoleEnum;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.enumeration.ServiceInstanceTypeEnum;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.util.AttrUtil;
import cn.me.alphamq.nameserver.cache.Container;
import cn.me.alphamq.nameserver.event.model.service.PullBrokerAddressEvent;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.nameserver.store.ServiceInstance;
import com.alibaba.fastjson2.JSON;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 拉取 broker 的 ip 地址的事件的监听器
 *
 * @author f
 */
public class PullBrokerIpListener implements Listener<PullBrokerAddressEvent> {

    @Override
    public void onReceive(PullBrokerAddressEvent event) throws Exception {
        BrokerRoleEnum role = BrokerRoleEnum.of(event.getRole());
        Map<String, ServiceInstance> serviceInstanceMap = Container.getServiceInstanceManager().getServiceInstanceMap();
        List<String> brokerAddressList = new ArrayList<>();
        for (String key : serviceInstanceMap.keySet()) {
            ServiceInstance instance = serviceInstanceMap.get(key);
            // 跳过非 broker 的服务实例
            if (ServiceInstanceTypeEnum.BROKER != instance.getServiceInstanceType()) {
                continue;
            }

            String brokerGroup = AttrUtil.getGroupFromAttrs(instance.getAttrs());
            if (!brokerGroup.equals(event.getBrokerGroup())) {
                continue;
            }

            String brokerRole = AttrUtil.getRoleFromAttrs(instance.getAttrs());
            String instanceAddress = instance.getIp() + ":" + instance.getPort();
            if (BrokerRoleEnum.MASTER == role && BrokerRoleEnum.MASTER.getCode().equals(brokerRole)) {
                brokerAddressList.add(instanceAddress);
            } else if (BrokerRoleEnum.SLAVE == role && BrokerRoleEnum.SLAVE.getCode().equals(brokerRole)) {
                brokerAddressList.add(instanceAddress);
            } else if (BrokerRoleEnum.SINGLE == role && BrokerRoleEnum.SINGLE.getCode().equals(brokerRole)) {
                brokerAddressList.add(instanceAddress);
            }
        }

        // 返回响应
        event.getCtx().writeAndFlush(new TcpMsg(EventCodeEnum.PULL_BROKER_ADDRESS_RESP,
                JSON.toJSONBytes(new PullBrokerAddressResp(event.getMsgId(), brokerAddressList))));
    }
}
